package com.hl233.framework.utils;

import com.hl233.framework.common.Constants;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
 * 适用于在消费者代码中直接控制消费起始位置。
 * @author 张振强15065206763
 * @create 2024-09-04
 */
public class ManualOffsetResetExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("group.id", "defaul_group");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());

        try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
            TopicPartition tp = new TopicPartition(Constants.HL233_DEVICE_MSG_TOPIC, 0);
            long targetOffset = 1; // 替换为目标偏移量

            consumer.assign(Collections.singletonList(tp));
            consumer.seek(tp, targetOffset);

//            while (true) {
//                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
//                // 处理记录...
//            }
        }
    }
}
